package com.letv.mq;

import java.net.URISyntaxException;

import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;

import org.apache.commons.lang.StringUtils;
import org.apache.qpid.client.AMQAnyDestination;
import org.apache.qpid.client.AMQConnectionFactory;
import org.apache.qpid.client.message.AbstractJMSMessage;
import org.apache.qpid.url.URLSyntaxException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.letv.common.Constants;
import com.letv.conf.ConfigFactory;
import com.letv.work.special.SpecialJob;

/**
 * 消息队列，用于增量更新索引/CBASE
 *
 */
public class MQ {
	private final static String MQ_URL =  ConfigFactory.getString("mq.url");// 供搜索引擎使用的交换机"/vrsAdvice";
	private final static String M_QUEUE_NAME =  ConfigFactory.getString("mq.requestQueue");// "so.M.queue"
	protected final static Logger log = LoggerFactory.getLogger(MQ.class);

    private final static int SEND_INTERVAL_TIME = 10 * 60 * 1000;

    // 异常尝试的次数
    private final static int EXCEPTION_MAX_NUM = 100;

    // 异常时等待时间 60秒
    private final static int EXCEPTION_LONG_WAIT_TIME = 5000;

	/**
	 * 获取MQ服务器的链接
	 */
	private static SessionFactory getSessionFactory() {
        AMQConnectionFactory connectionFactory = new AMQConnectionFactory();
        SessionFactory sessionFactory = null;
		try {
		    connectionFactory.setConnectionURLString(MQ_URL+"&sync_publish='all'&sync_ack=true");
	        sessionFactory = new SessionFactory();
	        sessionFactory.setConnectionFactory(connectionFactory);
			log.info("mq connect success");
		} catch ( URLSyntaxException e) {
			log.error(e.getMessage(), e);
		}
		return sessionFactory;
	}

	/**
	 * 接收各端消息并放到本地队列
	 */
	class MQRev implements Runnable {
		private SessionFactory sessionFactory;
		protected Logger logger = LoggerFactory.getLogger(MQRev.class);
		MessageConsumer consumer;
		Session session;

		protected MQRev(SessionFactory sessionFactory) {
			this.sessionFactory = sessionFactory;
		}

		/**
		 * 负责接收各端mq消息
		 */
		public void doRev() {
            int exceptionNum = 0;
            boolean success = false;
            while (true) {
			    try {
			        confirmConn();
                    success = true;
			        //				long i = 0;
					// 从消息队列里取到下一条消息
                    if (success) {
                        long mqid = System.currentTimeMillis();
                        AbstractJMSMessage msg = (AbstractJMSMessage)consumer.receive();
                        String jsonStr = new String(msg.toBodyString());
                        logger.info(mqid+",mq Consumer :" + jsonStr);
                        MQMessage message = new MQMessage();
                        JSONObject json = null;
                        JSONObject letvMMSMessage = null;
                        JSONObject properties = null;
                        JSONObject playPlatform = null;
                        int actionCode = 0;
                        String id = "";
                        String platform = "";
                        try {
                            json =  JSON.parseObject(jsonStr);
                            letvMMSMessage = json.getJSONObject("LetvMMSMessage");
                            properties = letvMMSMessage.getJSONObject("properties");
                            id = properties.getString("id");
                            actionCode = letvMMSMessage.getInteger("messageActionCode");
                        } catch (Exception e1) {
                            logger.warn(mqid+",parse the mq message error! continue!", e1);
                            continue;
                        }
                        try {
                            playPlatform = properties.getJSONObject("playPlatform");
                            platform = playPlatform.getString("new");
                        } catch (Exception e1) {
                            logger.warn(mqid+",parse the mq message 'playPlatform or new' error! go on...", e1);
                        }
                        message.setId(id);
                        if (101 <= actionCode && actionCode <= 103) {
                            message.setType(Constants.INDEX_ALBUM);
                        } else if (104 <= actionCode && actionCode <= 106) {
                            message.setType(Constants.INDEX_VIDEO);
                        } else if (107 <= actionCode && actionCode <= 109) {
                            message.setType(Constants.INDEX_STAR);
                        } else if (8881001 <= actionCode && actionCode <= 8881003) {
                            message.setType(Constants.INDEX_LESO);
                        } else if (8882001 <= actionCode && actionCode <= 8882003) {
                            message.setType(Constants.INDEX_LESO_EXTRAET);
                            actionCode += 2;
                        } else {
                            continue;
                        }

                        if (actionCode % 3 == 0) {			//修改
                            message.setAction(Constants.MQ_ACTION_MOD);
                        } else if (actionCode % 3 == 1) {	//删除
                            message.setAction(Constants.MQ_ACTION_DEL);
                        } else if (actionCode % 3 == 2) {	//添加
                            message.setAction(Constants.MQ_ACTION_ADD);
                        } else {
                            continue;
                        }

                        //专辑&视频播放平台为空时操作置为删除
                        if (101 <= actionCode && actionCode <= 106) {
                            if (StringUtils.isBlank(platform)) {
                                message.setAction(Constants.MQ_ACTION_DEL);
                                logger.info(mqid+",play_platform is " + playPlatform + ": type=" + message.getType() + ", id=" + message.getId() + ", DEL");
                            }
                        }

                        if (8882003 <= actionCode && actionCode <= 8882005 && actionCode % 3 == 1) {
                            message.setAction(letvMMSMessage.getString("redeliveryMode"));
                        }

                        if (message == null || StringUtils.isBlank(jsonStr)) {
                            continue;
                        }
                        // 调用增量
                        SpecialJob increme = new SpecialJob();
                        increme.fromMqMessage(mqid, message.getId(), message.getAction(), message.getType());
                        exceptionNum = 0;
                    }
			    } catch (Exception e) {
                    try {
                        // 如果出现异常，尝试次数超过一定数量后，进后长时间尝试
                        if(exceptionNum++ > EXCEPTION_MAX_NUM) {
                            // 尝试次数过多，进入长时间等待过程
                            sessionFactory = null;
                            session = null;
                            Thread.sleep(EXCEPTION_LONG_WAIT_TIME);
                        } else {
                            Thread.sleep(SEND_INTERVAL_TIME);
                        }
                    } catch (InterruptedException e1) {
                        logger.error("Call doSent thread sleep exception " + e1.getMessage());
                        break;
                    }
                    success = false;
                    logger.error("Connection or recieve Error! ", e);
			    }
            }
		}

		@Override
        public void run() {
			logger.info("rev thread starting.");
			doRev();
		}

	    public void confirmConn() throws JMSException, InterruptedException, URISyntaxException {
	        if ( sessionFactory == null ) {
	            this.sessionFactory = getSessionFactory();
	        }
	        if (session == null ) {
	            session = sessionFactory.createSession(false, Session.AUTO_ACKNOWLEDGE);
	            AMQAnyDestination queue = new AMQAnyDestination(M_QUEUE_NAME);
	            consumer = session.createConsumer(queue);
	        }
	    }

	}

	public static void start() {
        for (int i=0;i<8;i++) {
            SessionFactory sessionFactory  = getSessionFactory();// 建立MQ连接
            Thread revThread = new Thread(new MQ().new MQRev(sessionFactory));
            revThread.start();
        }
	}

    public static void main(String[] args) {
	}
}